/*
 * This file is a part of qloudgen-esb. 
 * You can redistribute qloudgen-esb and/or modify it under the terms of the Lesser GNU General Public License version 3. 
 * qloudgen-esb is distributed WITHOUT ANY WARRANTY. 
 * 
 * See the Lesser GNU General Public License for more details at http://www.gnu.org/licenses/.
 *  
 * Copyright (C) 2014 Qloudgen Technology
*/

package com.qloudgen.csb.test;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.log4j.PropertyConfigurator;

import com.qloudgen.csb.client.KfkConsumer;
import com.qloudgen.csb.client.KfkMessageListener;
import com.qloudgen.csb.client.KfkProducer;

public class Server5 extends ServerBase implements KfkMessageListener {  
	
	private KfkConsumer kc;
	private String topic;
	private KfkProducer kp;
	private String message;
	private ExecutorService executor;
	
	private Server5(String zkConnect, String topic, String groupId, int msgSize,
    		int bufferSize, String reply, int ackMode, int perfInterval) {
		super(perfInterval);
		kc = new KfkConsumer(reply, groupId,
				zkConnect, bufferSize, null);
		kc.setMessageListener(this);
		
		this.topic = topic;
		message = Utils.randomString(msgSize);
		kp = new KfkProducer(zkConnect, ackMode, null);
		executor = Executors.newFixedThreadPool(1);
	}
	
	@Override
	public void onMessage(String message, long offset) {
		count++;	

		//send next message
		final String msg = message;
		executor.submit(new Runnable() {
			@Override
			public void run() {
				kp.send(topic, msg);
			}			
		});
	}

	public void receive() {
		kp.send(topic, message);
		kc.run();
	}
	
    public static void main(String[] args) throws Exception {
    	PropertyConfigurator.configure("conf/log4j.properties");

    	final String zkConn = args[0];
    	final String topic = args[1];
    	final String gid = args[2];
    	final int msgSize = Integer.parseInt(args[3]);
    	final int bufferSize = Integer.parseInt(args[4]); 
    	final String reply = args[5];
    	final int ackMode = Integer.parseInt(args[6]);
    	final int nThreads = Integer.parseInt(args[7]);
    	final int perfInterval = Integer.parseInt(args[8]);

    	final Sum isum = new Sum(nThreads);    	
    	for (int i = 0; i < nThreads; i++) {
    		new Thread(new Runnable() {    		
    			@Override
    			public void run() {
    				Server5 srv = new  Server5(zkConn, topic, gid, msgSize,
    					bufferSize, reply, ackMode, perfInterval);
    				srv.perf(isum);
    	    		srv.receive();
    			}
    		}).start();
    	}

    	while (!Thread.currentThread().isInterrupted()) {
    		try {
    			System.out.println("server6 running with " 
    					+ nThreads + " thread...");
				Thread.sleep(Integer.MAX_VALUE);
			} catch (InterruptedException e) {}
    	}
    	System.out.println("server6 exiting...");
    	System.exit(0);
    }
}

